package com.atguigu.gmall.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.annotaion.NoSink;
import com.atguigu.gmall.bean.TableProcess;
import com.atguigu.gmall.bean.VisitorStats;
import com.atguigu.gmall.common.Constant;
import com.atguigu.gmall.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/28 9:21
 */
public class FlinkSinkUtil {
    
    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        Properties conf = new Properties();
        conf.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        conf.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, null, element.getBytes(StandardCharsets.UTF_8));
                }
            },
            conf,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
        
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties conf = new Properties();
        conf.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        conf.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                @Nullable Long timestamp) {
                    
                    return new ProducerRecord<>(element.f1.getSink_table(),
                                                null,
                                                element.f0.toJSONString().getBytes(StandardCharsets.UTF_8));
                }
            },
            conf,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getHbaseSink() {
        return new PhoenixSink();
    }
    
    public static void main(String[] args) {
        getClickHouseSink(Constant.CLICKHOUSE_DB, Constant.TABLE_VISITOR_STATS, VisitorStats.class);
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String db,
                                                        String table,
                                                        Class<T> tClass) {
        Field[] fields = tClass.getDeclaredFields();
        
        // TODO
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL_PRE + "/" + db;
        // insert into t(name, age, sex,) values(?, ?, ?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("insert into ")
            .append(table)
            .append("(");
        // 找到字段名  如果这个字段有NoSink这个注解,则不要拼接
        for (Field field : fields) {
            NoSink noSink = field.getAnnotation(NoSink.class);
            
            if (noSink == null) {
                sql
                    .append(field.getName())
                    .append(",");
            }
        }
        sql.deleteCharAt(sql.length() - 1); // 把最后一个逗号删除
        sql.append(") values(");
        // 拼接占位符
        for (Field field : fields) {
            NoSink noSink = field.getAnnotation(NoSink.class);
            if (noSink == null) {
                sql.append("?,");
            }
            
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(")");
        
        // System.out.println(sql.toString());
        
        // 借助jdbc sink封装一个ClickHouse sink
        return getJdbcSink(
            url,
            driver,
            sql.toString()
        );
    }
    
    public static <T> SinkFunction<T> getJdbcSink(String url,
                                                  String driver,
                                                  String sql) {
        return JdbcSink.sink(
            sql,
            new JdbcStatementBuilder<T>() {
                
                @Override
                public void accept(PreparedStatement ps,
                                   T t) throws SQLException {
                    System.out.println("t: " + t);
                    // 类名.class   对象.getClass()  Class.forName("...")
                    Class<?> tClass = t.getClass();
                    try {
                        Field[] fields = tClass.getDeclaredFields();
                        for (int i = 0, position = 1; i < fields.length; i++) {
                            
                            Field field = fields[i];
                            NoSink noSink = field.getAnnotation(NoSink.class);
                            if (noSink == null) {
                                field.setAccessible(true);
                                Object v = field.get(t);
                                ps.setObject(position++, v);
                            }
                        }
                        
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                    
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchIntervalMs(100)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(url)
                .withDriverName(driver)
                .build()
        );
    }
}
